package org.jpush.core;

import java.util.LinkedList;
import java.util.List;

/**
 * This is for handling all the events of a connection including
 * initialize/read(data come in)/write(data sent out).
 * See <code>PhysicalConnectionEventQueue.PhysicalConnectionEvent.EventType</code>
 * It will call the <code>Dispatcher</code> to handle
 * 
 * 
 * 
 * @author Lei Wang
 *
 */
public class Demultiplexor
{
	private Server server;
	private List<Thread> workerThreads;
	private boolean isRunning;
	
	public Demultiplexor(Server server)
	{
		this.server = server;
	}

	public boolean start()
	{
	    int nThreads = server.getWorkerCount();

	    workerThreads = new LinkedList<Thread>();
	    Thread thread = null;
	    this.isRunning = true;

	    for ( int i = 0; i < nThreads; i++ )
	    {

	    	thread = new Thread(new Runnable()
	    	{
				public void run()
				{
					// TODO Auto-generated method stub
					proc();
				}
	    		
	    	});
	    	
	    	// Add into the list
	    	workerThreads.add(thread);
	    	
	    	// Start the thread
	    	thread.start();
	    }
	    return true;
	}
	
	private void proc()
	{
		PhysicalConnectionEvent event = null;
		PhysicalConnectionEvent.EventType eventType = null;
		while (isRunning)
		{
			event = server.getConnectionEventQueue().getEvent();
			if (event != null)
			{
				eventType = event.getEventType();
				if (PhysicalConnectionEvent.EventType.read == eventType)
				{
					server.getDispatcher().onReceiveComplete(event.getConnection(), -1);
				}
				else if (PhysicalConnectionEvent.EventType.write == eventType)
				{
					server.getDispatcher().onWriteComplete(event.getConnection(), -1);
				}
				else if (PhysicalConnectionEvent.EventType.initialize == eventType)
				{
					server.getDispatcher().onInitializeReady(event.getConnection());
				}
				else
				{
					// Nothing will be done.
				}
			}
		}
	}
	
	private void waitAllWorkerThreads()
	{
		if (workerThreads == null)
		{
			return;
		}
		
		// Wait thread join
	    for (Thread t : workerThreads)
	    {
	    	try
	    	{
				t.join();
			}
	    	catch (InterruptedException e)
	    	{
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
	    }
	}
	
	private void destroyAllWorkerThreads()
	{
		if (workerThreads != null)
		{
			workerThreads.clear();
			workerThreads = null;
		}
	}

    public void stop()
    {
    	// Ask all worker threads to start shutting down
	    this.isRunning = false;

	    // Ask event queue to tell the worker thread to terminate
	    server.getConnectionEventQueue().postTerminateSignal();
	    
	    // Wait all the threads to terminate
	    waitAllWorkerThreads();
	    
	    // Destroy all the threads
	    destroyAllWorkerThreads();
    }
}
